package com.pig4cloud.pigx.contract.websocket.manager;

import com.pig4cloud.pigx.contract.vo.AbstractTopicVO;
import com.pig4cloud.pigx.contract.websocket.message.SubscribeMessage;
import com.pig4cloud.pigx.contract.websocket.message.UnSubscribeMessage;
import com.pig4cloud.pigx.contract.websocket.subscribe.CurrentPosition;
import com.pig4cloud.pigx.contract.websocket.subscribe.DepthMap;
import com.pig4cloud.pigx.contract.websocket.subscribe.Kline;
import com.pig4cloud.pigx.contract.websocket.subscribe.MergeContract;
import com.pig4cloud.pigx.contract.websocket.subscribe.MergeKLine;
import com.pig4cloud.pigx.contract.websocket.subscribe.OptionalSymbol;
import com.pig4cloud.pigx.contract.websocket.subscribe.Ranking;
import com.pig4cloud.pigx.contract.websocket.subscribe.SymbolRiseAndFall;
import jodd.util.collection.MapEntry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class WsManager {

	@Autowired
	public Kline kline;

	@Autowired
	public Ranking ranking;

	@Autowired
	public OptionalSymbol optionalSymbol;

	@Autowired
	public SymbolRiseAndFall riseAndFall;

	@Autowired
	public CurrentPosition currentPosition;

	@Autowired
	public DepthMap depthMap;

	@Autowired
	public MergeContract mergeContract;

	@Autowired
	public MergeKLine mergeKLine;

	public Map<WebSocketSession, Long> webSocketSessionMap = new ConcurrentHashMap<WebSocketSession, Long>();

	private WsManager() {
		init();
	}

	public void init() {
		check();
	}

	public void subscribe(WebSocketSession session, SubscribeMessage subscribeMessage) {
		if (webSocketSessionMap.get(session) == null) {
			webSocketSessionMap.put(session, System.currentTimeMillis());
		}

		if (subscribeMessage.getParams().size() > 0 && !subscribeMessage.getParams().get(0).isEmpty()) {
			String topicName = subscribeMessage.getParams().get(0);
			switch (topicName) {
				case "Kline":
					kline.sub(session, subscribeMessage.getParams());
					break;
				case "ranking":
					ranking.sub(session, subscribeMessage.getParams());
					AbstractTopicVO sub = ranking.sub(session, subscribeMessage.getParams());
					Map.Entry<WebSocketSession, AbstractTopicVO> entry = new MapEntry<>(session, sub);
					ranking.sendMessage(entry);
					break;
				case "optionalSymbol":
					AbstractTopicVO optionSymbolSub = optionalSymbol.sub(session, subscribeMessage.getParams());
					Map.Entry<WebSocketSession, AbstractTopicVO> optionSymbolEntry = new MapEntry<>(session, optionSymbolSub);
					optionalSymbol.sendMessage(optionSymbolEntry);
					break;
				case "riseAndFall":
					riseAndFall.sub(session, subscribeMessage.getParams());
					break;
				case "currentPosition":
					currentPosition.sub(session, subscribeMessage.getParams());
					break;
				case "depthMap":
					depthMap.sub(session, subscribeMessage.getParams());
					break;
				case "mergeContract":
					AbstractTopicVO mergeSub = mergeContract.sub(session, subscribeMessage.getParams());
					Map.Entry<WebSocketSession, AbstractTopicVO> mergeEntry = new MapEntry<>(session, mergeSub);
					mergeContract.sendMessage(mergeEntry);
					break;
				case "mergeKLine":
					mergeKLine.sub(session, subscribeMessage.getParams());
					break;
				default:
					break;
			}
		}
	}

	public void unsubscribe(WebSocketSession session, UnSubscribeMessage unsubscribeMessage) {
		if (unsubscribeMessage.getParams().size() > 0 && !unsubscribeMessage.getParams().get(0).isEmpty()) {
			String topicName = unsubscribeMessage.getParams().get(0);
			switch (topicName) {
				case "Kline":
					kline.unsub(session, unsubscribeMessage.getParams());
					break;
				case "ranking":
					ranking.unsub(session, unsubscribeMessage.getParams());
					break;
				case "optionalSymbol":
					optionalSymbol.unsub(session, unsubscribeMessage.getParams());
					break;
				case "riseAndFall":
					riseAndFall.unsub(session, unsubscribeMessage.getParams());
					break;
				case "currentPosition":
					currentPosition.unsub(session, unsubscribeMessage.getParams());
					break;
				case "depthMap":
					depthMap.unsub(session, unsubscribeMessage.getParams());
					break;
				case "mergeContract":
					mergeContract.unsub(session, unsubscribeMessage.getParams());
					break;
				case "mergeKLine":
					mergeKLine.unsub(session, unsubscribeMessage.getParams());
					break;
				default:
					break;
			}
		}
	}

	private void check() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
		service.scheduleAtFixedRate(() -> {
			try {
				webSocketSessionMap.forEach((k, v) -> {
					if (!k.isOpen()) {
						webSocketSessionMap.remove(k);
						if (kline.subList.containsKey(k)) {
							kline.subList.remove(k);
						}
						if (ranking.subList.containsKey(k)) {
							ranking.subList.remove(k);
						}
						if (optionalSymbol.subList.containsKey(k)) {
							optionalSymbol.subList.remove(k);
						}
						if (riseAndFall.subList.containsKey(k)) {
							riseAndFall.subList.remove(k);
						}
						if (currentPosition.subList.containsKey(k)) {
							currentPosition.subList.remove(k);
						}
						if (depthMap.subList.containsKey(k)) {
							depthMap.subList.remove(k);
						}
						if (mergeContract.subList.containsKey(k)) {
							mergeContract.subList.remove(k);
						}
						if (mergeKLine.subList.containsKey(k)) {
							mergeKLine.subList.remove(k);
						}
					}
				});
			} catch (Exception ex) {

			}
		}, 10, 50, TimeUnit.SECONDS);

	}

}
